package com.shujia.spark.stream

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}

import java.lang
import java.sql.{Connection, Date, DriverManager, PreparedStatement}
import java.text.SimpleDateFormat

object Demo7Card {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .master("local[2]")
      .appName("card")
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    /**
     * 实时统计卡口拥堵情况
     * 1、统计最近一段时间内的平均车速和车流量
     */

    /**
     * 1、读取卡口过车数据
     */
    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    /**
     * 2、解析json格式的数据，取出卡口编号和车速
     */

    val kvDS: DStream[(Long, (Double, Int))] = linesDS.map((line: String) => {
      //将json字符串转换成json对象
      val jsonObj: JSONObject = JSON.parseObject(line)
      //卡口编号
      val card: Long = jsonObj.getLong("card")
      //车速
      val speed: Double = jsonObj.getDouble("speed")
      (card, (speed, 1))
    })

    /**
     * 3、统计最近一段时间内总的车速和总的车流量
     */
    val sumFlowAndSumSpeed: DStream[(Long, (Double, Int))] = kvDS.reduceByKeyAndWindow(
      (kv1: (Double, Int), kv2: (Double, Int)) => (kv1._1 + kv2._1, kv1._2 + kv2._2),
      Durations.seconds(15),
      Durations.seconds(5)
    )

    /**
     * 4、计算平均车速
     */
    val sumFlowAndAvgSpeed: DStream[(Long, Int, Double)] = sumFlowAndSumSpeed.map {
      case (card: Long, (speed: Double, flow: Int)) =>
        //平均车速
        val avgSpeed: Double = speed / flow
        (card, flow, avgSpeed)
    }


    /**
     * 5、将统计结果保存到mysql中
     */
    sumFlowAndAvgSpeed.foreachRDD(rdd => {

      rdd.foreachPartition(iter => {
        //创建数据库链接

        Class.forName("com.mysql.jdbc.Driver")
        val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student","root","123456")
        val stat: PreparedStatement = con.prepareStatement("insert into card_flow_avg_speed(card,com_time,flow,avg_speed) values(?,?,?,?)")

        //时间取当前计算时间
        //获取当前的时间戳
        val ts: Long = System.currentTimeMillis()
        //将时间戳转换成日期对象
        val date = new Date(ts)
        //构建时间格式化对象
        val format = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss")
        //将日期对象转换成时间字符串
        val comTime: String = format.format(date)

        iter.foreach {
          case (card: Long, flow: Int, avgSpeed: Double) =>
            //指定字段
            stat.setLong(1, card)
            stat.setString(2, comTime)
            stat.setInt(3, flow)
            stat.setDouble(4, avgSpeed)
            stat.execute()
        }

        stat.close()
        con.close()

      })
    })


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
